package com.niit.spark.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/22
 * Author：Ys
 * Description:
 */
object StreamingKafka03 {

  def main(args: Array[String]): Unit = {
    val ssc = new StreamingContext(new SparkConf().setMaster("local[*]").setAppName("StreamingKafka03"), Seconds(5))
    ssc.sparkContext.setLogLevel("ERROR")

    //创建Kafka的配置信息
    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "SP_KF",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName
    )

    //消费Kafka中的数据 创建DStream
    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent, //官方推荐的本地策略
      ConsumerStrategies.Subscribe[String,String](Set("BD2"), kafkaParams)
    )

    val infoDS: DStream[String] = kafkaDS.map(record => {
      val topic: String = record.topic()
      val partition: Int = record.partition()
      val offset: Long = record.offset()
      val key: String = record.key()
      val value: String = record.value()
      val info = s"topic: $topic, partition: $partition, offset: $offset, key: $key, value: $value"
      info
    })
    infoDS.print()


    ssc.start()
    ssc.awaitTermination()

  }

}
